package com.im.dispatcher;

import com.alibaba.fastjson.JSONObject;
import com.google.protobuf.InvalidProtocolBufferException;
import com.im.common.*;
import com.im.protocol.*;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandler;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.socket.SocketChannel;
import io.netty.util.concurrent.EventExecutorGroup;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * @Created leijun
 * @Date 2021/9/12 8:24 下午
 */
public class DispatcherHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        GatewayInstanceManager gatewayInstanceManager = GatewayInstanceManager.getInstance();
        gatewayInstanceManager.remove((SocketChannel) ctx.channel());
    }

    /**
     * 与接入层建立连接调用此方法
     * @param ctx
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        GatewayInstanceManager gatewayInstanceManager = GatewayInstanceManager.getInstance();
        gatewayInstanceManager.addInstance((SocketChannel) ctx.channel());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Message message = new Message((ByteBuf) msg);
        if (message.getMessageType() == ImConstant.MESSAGE_TYPE_REQUEST) {
            Request request = message.toRequest();
            // 如果是用户认证请求，向SSO发起认证，认证成功缓存sessionaao
            if (request.getRequestType() == ImConstant.REQUEST_TYPE_AUTHENTICATE) {
                handleAuthenticateRequest(ctx, request);
            } else if (request.getRequestType() == ImConstant.REQUEST_TYPE_SEND_MESSAGE) {
                handleSendMessageRequest(ctx, request);
            } else if (request.getRequestType() == ImConstant.REQUEST_TYPE_GROUP_MESSAGE) {
                handleSendGroupMessageRequest(ctx, request);
            }
        } else if (message.getMessageType() == ImConstant.MESSSAGE_TYPE_RESPONSE) {
            Response response = message.toResponse();
            if (response.getRequestType() == ImConstant.REQUEST_TYPE_PUSH_MESSAGE) {
                handlePushMessageResponse(ctx, response);
            }
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        super.channelReadComplete(ctx);
    }

    /**
     * 处理接入层的用户认证请求
     * @param ctx
     * @param request
     */
    public void handleAuthenticateRequest(ChannelHandlerContext ctx, Request request) throws InvalidProtocolBufferException {
        SocketChannel socketChannel = (SocketChannel) ctx.channel();
        RequestHandler requestHandler = RequestHandler.getInstance();
        AuthenticateRequestProto.AuthenticateRequest authenticateRequest =
                AuthenticateRequestProto.AuthenticateRequest.parseFrom(request.getBody());
        AuthenticateResponseProto.AuthenticateResponse authenticateResponse =
                requestHandler.authenticate(authenticateRequest);
        if (authenticateResponse.getStatus() == ResponseCode.OK) {
            String socketChannelId = socketChannel.remoteAddress().getHostName() + ":" + socketChannel.remoteAddress().getPort();
            String sessionKey = "session_"+authenticateResponse.getUid();
            SessionInfo sessionInfo = new SessionInfo();
            sessionInfo.setToken(authenticateResponse.getToken());
            sessionInfo.setTimestamp(authenticateResponse.getTimestamp());
            sessionInfo.setAuthTimestamp(System.currentTimeMillis());
            sessionInfo.setChannelId(socketChannelId);
            JedisManager jedisManager = JedisManager.getInstance();
            jedisManager.getJedis().set(sessionKey, JSONObject.toJSONString(sessionInfo));
        }
        Response response = new Response(request.getAppSdkVersion(), request.getRequestType(),
                request.getSequence(), authenticateResponse.toByteArray());
        ctx.writeAndFlush(response.getBuf());
    }

    /**
     * 处理接入层的单聊消息请求
     * @param ctx
     * @param request
     * @throws InvalidProtocolBufferException
     */
    public void handleSendMessageRequest(ChannelHandlerContext ctx, Request request) throws InvalidProtocolBufferException {
        SocketChannel socketChannel = (SocketChannel) ctx.channel();
        MessageRequestProto.MessageRequest messageRequest =
                MessageRequestProto.MessageRequest.parseFrom(request.getBody());

        JSONObject value = new JSONObject();
        value.put("senderId", messageRequest.getSenderId());
        value.put("receiverId", messageRequest.getReceiverId());
        value.put("content", messageRequest.getContent());
        value.put("timestamp", System.currentTimeMillis());
        value.put("requestType", ImConstant.REQUEST_TYPE_SEND_MESSAGE);
        value.put("messageType", request.getMessageType());
        value.put("sequence", messageRequest.getSequence());
        String socketChannelId = socketChannel.remoteAddress().getHostName() + ":" + socketChannel.remoteAddress().getPort();
        System.out.println("【分发层】接收到客户端【"+socketChannelId+"】的单聊消息");

        KafkaManager kafkaManager = KafkaManager.getInstance();
        KafkaProducer producer = kafkaManager.getProducer();
        ProducerRecord<String, String> record = new ProducerRecord<>("single_message", value.toJSONString());
        producer.send(record);
         System.out.println("将消息【"+value.toJSONString()+"】推送到kafka");
    }

    /**
     * 发送群聊消息
     * @param ctx
     * @param request
     */
    private void handleSendGroupMessageRequest(ChannelHandlerContext ctx, Request request) throws InvalidProtocolBufferException {
        SocketChannel socketChannel = (SocketChannel) ctx.channel();
        GroupMessageRequestProto.GroupMessageRequest messageRequest =
                GroupMessageRequestProto.GroupMessageRequest.parseFrom(request.getBody());
        JSONObject value = new JSONObject();
        value.put("senderId", messageRequest.getSenderId());
        value.put("groupId", messageRequest.getGroupId());
        value.put("content", messageRequest.getContent());
        value.put("timestamp", System.currentTimeMillis());
        value.put("requestType", ImConstant.REQUEST_TYPE_GROUP_MESSAGE);
        value.put("messageType", request.getMessageType());
        value.put("sequence", ImConstant.SEQUENCE_DEFAULT);
        String socketChannelId = socketChannel.remoteAddress().getHostName() + ":" + socketChannel.remoteAddress().getPort();
        System.out.println("【分发层】接收到客户端【"+socketChannelId+"】的群聊消息");

        KafkaManager kafkaManager = KafkaManager.getInstance();
        KafkaProducer producer = kafkaManager.getProducer();
        ProducerRecord<String, String> record = new ProducerRecord<>("group_message", value.toJSONString());
        producer.send(record);
        System.out.println("将群聊消息【"+value.toJSONString()+"】推送到kafka");
    }

    /**
     * 处理推送响应消息
     * @param ctx
     * @param response
     */
    private void handlePushMessageResponse(ChannelHandlerContext ctx, Response response) throws InvalidProtocolBufferException {
        PushMessageResponseProto.PushMessageResponse pushMessageResponse =
                PushMessageResponseProto.PushMessageResponse.parseFrom(response.getBody());
        KafkaManager kafkaManager = KafkaManager.getInstance();
        KafkaProducer producer = kafkaManager.getProducer();
        JSONObject message = new JSONObject();
        message.put("messageId", pushMessageResponse.getMessageId());
        ProducerRecord<String, String> record = new ProducerRecord<String, String>("push_message_response", message.toJSONString());
        producer.send(record);
    }
}
